package cn.doitedu.rtmk.demo4;

import cn.doitedu.rtmk.beans.RuleMetaBean;
import cn.doitedu.rtmk.beans.UserEvent;
import cn.doitedu.rtmk.demo3.RuleEngineCoreFunction3;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
/**
 * @Author: deep as the sea
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/4/15
 * @Desc: 学大数据，到多易教育
 *   此版本：提炼 ==> 封装规则的计算逻辑类，并能支持多个规则
 **/
public class Demo4 {

    public static void main(String[] args) throws Exception {

        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        env.setParallelism(1);
        env.setStateBackend(new HashMapStateBackend());

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 从kafka中读取用户的实时行为
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setTopics("dwd-events-detail")
                .setClientIdPrefix("rtmk-")
                .setGroupId("g-rtmk-")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        // 用mysql-cdc去监听规则管理平台的mysql中 规则 元数据表，得到规则的预圈选人群
        tenv.executeSql(
                "CREATE TABLE rule_meta_mysql (    " +
                        "      rule_id BIGINT," +
                        "      pre_select_users BYTES,                " +
                        "     PRIMARY KEY (rule_id) NOT ENFORCED      " +
                        "     ) WITH (                                 " +
                        "     'connector' = 'mysql-cdc',               " +
                        "     'hostname' = 'doitedu'   ,               " +
                        "     'port' = '3306'          ,               " +
                        "     'username' = 'root'      ,               " +
                        "     'password' = 'root'      ,               " +
                        "     'database-name' = 'doit37',              " +
                        "     'table-name' = 'rule_meta'               " +
                        ")"
        );

        // 表转流，并将row转成我们自己的封装bean
        DataStream<Row> ruleMetaRowStream = tenv.toChangelogStream(tenv.from("rule_meta_mysql"));
        SingleOutputStreamOperator<RuleMetaBean> ruleMetaBeanStream = ruleMetaRowStream.map(new MapFunction<Row, RuleMetaBean>() {
            @Override
            public RuleMetaBean map(Row row) throws Exception {

                RuleMetaBean ruleMetaBean = new RuleMetaBean();
                ruleMetaBean.setRowKind(row.getKind().toByteValue());

                long rule_id = row.getFieldAs("rule_id");
                ruleMetaBean.setRule_id(rule_id);

                byte[] pre_select_users = row.getFieldAs("pre_select_users");
                ruleMetaBean.setPre_select_users(pre_select_users);

                return ruleMetaBean;
            }
        });

        // 广播 规则元信息 流
        MapStateDescriptor<String, String> bstDesc = new MapStateDescriptor<>("bst", String.class, String.class);
        BroadcastStream<RuleMetaBean> ruleMetaBeanBroadcastStream = ruleMetaBeanStream.broadcast(bstDesc);


        /**
         * 读取kafka中的用户行为事件
         */
        DataStreamSource<String> eventStrStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s");
        SingleOutputStreamOperator<UserEvent> eventStream = eventStrStream.map(json -> JSON.parseObject(json, UserEvent.class));

        SingleOutputStreamOperator<JSONObject> resultStream =
                eventStream
                        .keyBy(UserEvent::getUser_id)
                        // 连接元信息广播流  和  用户事件主流
                        .connect(ruleMetaBeanBroadcastStream)
                        .process(new RuleEngineCoreFunction4());

        resultStream.print();

        env.execute();
    }

}
